package com.shujia.flink.soure

import org.apache.flink.streaming.api.scala._

object Demo1ListSourvce {
  def main(args: Array[String]): Unit = {

    //创建flink的运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //基于本地集合构建source 构建出来的流是一个有界流

    val ds: DataStream[List[Int]] = env.fromElements(List(1, 2, 3, 4, 5, 6, 7, 8, 9))

    ds.print()




    //基于文件构架流

    //有界流
    val students: DataStream[String] = env.readTextFile("flink/data/students.txt")


    val kvDS: DataStream[(String, Int)] = students.map(line => (line.split(",")(4), 1))

    //统计班级的人数
    val sumClazz: DataStream[(String, Int)] = kvDS.keyBy(_._1).sum(1)

    sumClazz.print()


    //基于socket构建流
    //无界流
    val socketDs: DataStream[String] = env.socketTextStream("master", 8888)


    socketDs.print()


    env.execute()


  }

}
